//异步工作器的实现

#ifndef __M_WORKER_H__
#define __M_WORKER_H__

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>

namespace mylog
{
    using Functor = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,//安全状态，表示缓冲区满了就阻塞，避免资源耗尽的风险
        ASYNC_UNSAFE//非安全状态，不考虑资源耗尽的问题--无限扩容，用于测试极限性能
    };

    class AsyncWorker
    {
    public:
        using ptr = std::shared_ptr<AsyncWorker>;
        AsyncWorker(const Functor &cb, AsyncType worker_type = AsyncType::ASYNC_SAFE):
            _stop(false),
            _worker_type(worker_type),
            _thread(std::thread(&AsyncWorker::threadEntry, this)),
            _call_Back(cb){}
        
        ~AsyncWorker() { stop(); }

        void stop()
        {
            _stop = true;//将退出标志设置为true
            _cond_con.notify_all();//唤醒所有工作线程（消费者）
            _thread.join();
        }
        void push(const char* data, size_t len)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            //安全模式下的条件变量控制，若生产者缓冲区的剩余可写空间大小大于数据长度，则可以添加数据
            if(_worker_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&](){ return _pro_buf.writeableSize() >= len; });
            //能够走下来代表满足条件，可以向缓冲区写入数据
            _pro_buf.push(data, len);
            //唤醒消费者对缓冲区的数据进行处理
            _cond_con.notify_one();
        }
    private:
        //线程入口函数--对缓冲区的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
        void threadEntry()
        {
            while(1)
            {
                //用{}为互斥锁设置一生命周期，当缓冲区交换完毕后就解锁（并不对数据的处理过程加锁保护）
                {
                    //判断生产者缓冲区是否有数据，有则交换，无则阻塞
                    std::unique_lock<std::mutex> lock(_mutex);
                    //如果退出标志被设置为true，并且生产缓冲区已空，才能退出，否则可能出现缓冲区有数据还未处理就退出的情况
                    if(_stop && _pro_buf.empty()) break;
                    //若当前是退出前被唤醒或者有数据被唤醒，则返回真往下运行，否则重新进入休眠
                    _cond_con.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);
                    //判断是否为安全模式，是的话则需要唤醒生产者
                    if(_worker_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                //对消费者缓冲区进行处理
                _call_Back(_con_buf);
                //初始化消费者缓冲区
                _con_buf.reset();
            }
        }
        Functor _call_Back;//具体对缓冲区数据进行处理的回调函数，由异步工作器使用者传入
    private:
        AsyncType _worker_type;
        std::atomic<bool> _stop;//工作器停止标志
        Buffer _pro_buf;//生产缓冲区
        Buffer _con_buf;//消费缓冲区
        std::mutex _mutex;
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;
        std::thread _thread;//异步工作器对应的工作线程
    };
}









#endif 
